package com.zheng.cms.rocketmq.consumer;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;

import com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.log4j.Logger;

public class OrderConsumer {

    private Logger log = Logger.getLogger(getClass());

    private DefaultMQPushConsumer consumer;

    private String consumerGroup;

    private String namesrvAddr;

    private String instanceName;

    private String topic;


    private MessageListenerOrderly messageListenerOrderly;

    /**
     * 设置一个Listener，主要进行消息的逻辑处理
     *注意这里使用的是MessageListenerOrderly这个接口
     * */
    public OrderConsumer(String consumerGroup, String namesrvAddr, String instanceName, String topic,
                         MessageListenerOrderly messageListenerOrderly) {
        this.consumerGroup = consumerGroup;
        this.namesrvAddr = namesrvAddr;
        this.instanceName = instanceName;
        this.topic = topic;
        this.messageListenerOrderly = messageListenerOrderly;
    }
/**
 * //这里设置的是一个consumer的消费策略
 *         //CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
 *         //CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
 *         //CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
 *         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
 *
 *         设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
 *          * 如果非第一次启动，那么按照上次消费的位置继续消费
 *   consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
 * */
    public void init() throws MQClientException {
        log.info("start init DefaultMQPushConsumer...");
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); //从队列头部开始消费
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setInstanceName(instanceName);
        consumer.subscribe(topic, "*");//订阅所有
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.registerMessageListener(messageListenerOrderly);
        consumer.start();
        log.info("DefaultMQPushConsumer init ok.");
    }

    public void destroy(){
        log.info("start destroy DefaultMQPushConsumer...");
        consumer.shutdown();
        log.info("DefaultMQPushConsumer destroy success.");
    }

    public DefaultMQPushConsumer getConsumer() {
        return consumer;
    }
}
